//! MongoDB数据源插件
//! 
//! 提供MongoDB数据库的连接和查询功能，支持SQL映射

use std::collections::HashMap;
use async_trait::async_trait;
use serde_json::{Value, json};
use anyhow::{Result, anyhow};
use mongodb::{Client, Database, Collection, bson::doc};
use mongodb::bson::{Document, Bson};

use crate::{DataSource, DataSourceQuery, DataSourceResult, DataSourceStats};

/// MongoDB配置
#[derive(Debug, Clone)]
pub struct MongoDBConfig {
    /// 连接字符串
    pub connection_string: String,
    /// 数据库名称
    pub database_name: String,
    /// 连接超时时间（秒）
    pub connect_timeout_seconds: u64,
    /// 查询超时时间（秒）
    pub query_timeout_seconds: u64,
    /// 最大连接数
    pub max_connections: u32,
}

impl Default for MongoDBConfig {
    fn default() -> Self {
        Self {
            connection_string: "mongodb://localhost:27017".to_string(),
            database_name: "test".to_string(),
            connect_timeout_seconds: 30,
            query_timeout_seconds: 60,
            max_connections: 10,
        }
    }
}

/// MongoDB数据源
pub struct MongoDBDataSource {
    /// 配置
    config: MongoDBConfig,
    /// MongoDB客户端
    client: Option<Client>,
    /// 数据库实例
    database: Option<Database>,
    /// 连接状态
    connected: bool,
}

impl MongoDBDataSource {
    /// 创建新的MongoDB数据源
    pub fn new(config: MongoDBConfig) -> Self {
        Self {
            config,
            client: None,
            database: None,
            connected: false,
        }
    }

    /// 解析SQL查询为MongoDB操作
    fn parse_sql_to_mongodb(&self, sql: &str) -> Result<MongoDBOperation> {
        let sql = sql.trim().to_lowercase();
        
        if sql.starts_with("select") {
            self.parse_select_query(&sql)
        } else if sql.starts_with("insert") {
            self.parse_insert_query(&sql)
        } else if sql.starts_with("update") {
            self.parse_update_query(&sql)
        } else if sql.starts_with("delete") {
            self.parse_delete_query(&sql)
        } else {
            Err(anyhow!("Unsupported SQL operation: {}", sql))
        }
    }

    /// 解析SELECT查询
    fn parse_select_query(&self, sql: &str) -> Result<MongoDBOperation> {
        // 简化的SQL解析，实际应该使用更完整的SQL解析器
        if sql.contains("from") {
            let parts: Vec<&str> = sql.split_whitespace().collect();
            let mut collection_name = "default";
            let mut filter = doc! {};
            let mut limit = None;
            
            // 查找表名
            if let Some(from_index) = parts.iter().position(|&x| x == "from") {
                if from_index + 1 < parts.len() {
                    collection_name = parts[from_index + 1];
                }
            }
            
            // 查找WHERE条件
            if sql.contains("where") {
                // 简化处理：只支持基本的等值查询
                if let Some(where_part) = sql.split("where").nth(1) {
                    let where_part = where_part.split("limit").next().unwrap_or(where_part).trim();
                    if let Some((field, value)) = self.parse_simple_condition(where_part) {
                        filter.insert(field, value);
                    }
                }
            }
            
            // 查找LIMIT
            if sql.contains("limit") {
                if let Some(limit_part) = sql.split("limit").nth(1) {
                    if let Ok(limit_value) = limit_part.trim().parse::<i64>() {
                        limit = Some(limit_value);
                    }
                }
            }
            
            Ok(MongoDBOperation::Find {
                collection: collection_name.to_string(),
                filter,
                limit,
            })
        } else {
            Err(anyhow!("Invalid SELECT query"))
        }
    }

    /// 解析简单的WHERE条件
    fn parse_simple_condition(&self, condition: &str) -> Option<(String, Bson)> {
        if condition.contains("=") {
            let parts: Vec<&str> = condition.split("=").collect();
            if parts.len() == 2 {
                let field = parts[0].trim().to_string();
                let value = parts[1].trim();
                
                // 尝试解析为不同类型
                if let Ok(int_val) = value.parse::<i32>() {
                    return Some((field, Bson::Int32(int_val)));
                } else if let Ok(int64_val) = value.parse::<i64>() {
                    return Some((field, Bson::Int64(int64_val)));
                } else if let Ok(float_val) = value.parse::<f64>() {
                    return Some((field, Bson::Double(float_val)));
                } else if value.starts_with("'") && value.ends_with("'") {
                    let string_val = &value[1..value.len()-1];
                    return Some((field, Bson::String(string_val.to_string())));
                } else if value.starts_with("\"") && value.ends_with("\"") {
                    let string_val = &value[1..value.len()-1];
                    return Some((field, Bson::String(string_val.to_string())));
                } else {
                    return Some((field, Bson::String(value.to_string())));
                }
            }
        }
        None
    }

    /// 解析INSERT查询
    fn parse_insert_query(&self, _sql: &str) -> Result<MongoDBOperation> {
        // 简化实现
        Ok(MongoDBOperation::Insert {
            collection: "default".to_string(),
            document: doc! {"message": "Insert operation not fully implemented"},
        })
    }

    /// 解析UPDATE查询
    fn parse_update_query(&self, _sql: &str) -> Result<MongoDBOperation> {
        // 简化实现
        Ok(MongoDBOperation::Update {
            collection: "default".to_string(),
            filter: doc! {},
            update: doc! {"$set": {"updated": true}},
        })
    }

    /// 解析DELETE查询
    fn parse_delete_query(&self, _sql: &str) -> Result<MongoDBOperation> {
        // 简化实现
        Ok(MongoDBOperation::Delete {
            collection: "default".to_string(),
            filter: doc! {},
        })
    }

    /// 执行MongoDB操作
    async fn execute_mongodb_operation(&self, operation: MongoDBOperation) -> Result<Value> {
        let database = self.database.as_ref()
            .ok_or_else(|| anyhow!("Database not connected"))?;

        match operation {
            MongoDBOperation::Find { collection, filter, limit } => {
                let coll: Collection<Document> = database.collection(&collection);
                
                let mut cursor = if let Some(limit_val) = limit {
                    coll.find(filter).limit(limit_val).await?
                } else {
                    coll.find(filter).await?
                };

                let mut results = Vec::new();
                while cursor.advance().await? {
                    let doc = cursor.current();
                    // 将RawDocument转换为Document
                    let document: Document = doc.try_into()?;
                    let json_value = self.bson_document_to_json(&document)?;
                    results.push(json_value);
                }

                Ok(json!(results))
            }
            MongoDBOperation::Insert { collection, document } => {
                let coll: Collection<Document> = database.collection(&collection);
                let result = coll.insert_one(document).await?;
                Ok(json!({
                    "inserted_id": result.inserted_id.to_string()
                }))
            }
            MongoDBOperation::Update { collection, filter, update } => {
                let coll: Collection<Document> = database.collection(&collection);
                let result = coll.update_many(filter, update).await?;
                Ok(json!({
                    "matched_count": result.matched_count,
                    "modified_count": result.modified_count
                }))
            }
            MongoDBOperation::Delete { collection, filter } => {
                let coll: Collection<Document> = database.collection(&collection);
                let result = coll.delete_many(filter).await?;
                Ok(json!({
                    "deleted_count": result.deleted_count
                }))
            }
        }
    }

    /// 将BSON文档转换为JSON
    fn bson_document_to_json(&self, doc: &Document) -> Result<Value> {
        let mut json_obj = serde_json::Map::new();
        
        for (key, value) in doc {
            let json_value = self.bson_to_json(value)?;
            json_obj.insert(key.clone(), json_value);
        }
        
        Ok(Value::Object(json_obj))
    }

    /// 将BSON值转换为JSON值
    fn bson_to_json(&self, bson: &Bson) -> Result<Value> {
        match bson {
            Bson::Double(f) => Ok(json!(f)),
            Bson::String(s) => Ok(json!(s)),
            Bson::Array(arr) => {
                let mut json_arr = Vec::new();
                for item in arr {
                    json_arr.push(self.bson_to_json(item)?);
                }
                Ok(Value::Array(json_arr))
            }
            Bson::Document(doc) => self.bson_document_to_json(doc),
            Bson::Boolean(b) => Ok(json!(b)),
            Bson::Null => Ok(Value::Null),
            Bson::Int32(i) => Ok(json!(i)),
            Bson::Int64(i) => Ok(json!(i)),
            Bson::ObjectId(oid) => Ok(json!(oid.to_string())),
            Bson::DateTime(dt) => Ok(json!(dt.to_string())),
            _ => Ok(json!(bson.to_string())),
        }
    }
}

/// MongoDB操作类型
#[derive(Debug, Clone)]
enum MongoDBOperation {
    Find {
        collection: String,
        filter: Document,
        limit: Option<i64>,
    },
    Insert {
        collection: String,
        document: Document,
    },
    Update {
        collection: String,
        filter: Document,
        update: Document,
    },
    Delete {
        collection: String,
        filter: Document,
    },
}

#[async_trait]
impl DataSource for MongoDBDataSource {
    fn name(&self) -> &str {
        "MongoDB"
    }

    fn description(&self) -> &str {
        "MongoDB文档数据库数据源"
    }

    async fn connect(&mut self) -> Result<()> {
        tracing::info!("连接到MongoDB数据库: {}", self.config.connection_string);
        
        let client = Client::with_uri_str(&self.config.connection_string).await?;
        let database = client.database(&self.config.database_name);
        
        // 测试连接
        database.run_command(doc! {"ping": 1}).await?;
        
        self.client = Some(client);
        self.database = Some(database);
        self.connected = true;
        
        tracing::info!("MongoDB连接成功");
        Ok(())
    }

    async fn disconnect(&mut self) -> Result<()> {
        self.client = None;
        self.database = None;
        self.connected = false;
        tracing::info!("MongoDB连接已断开");
        Ok(())
    }

    async fn is_connected(&self) -> bool {
        self.connected
    }

    async fn execute_query(&mut self, query: DataSourceQuery) -> Result<DataSourceResult> {
        let start_time = std::time::Instant::now();
        
        if !self.connected {
            return Ok(DataSourceResult {
                success: false,
                data: Value::Null,
                error: Some("MongoDB not connected".to_string()),
                stats: DataSourceStats {
                    execution_time_ms: 0.0,
                    rows_processed: 0,
                    bytes_processed: 0,
                },
            });
        }

        match self.parse_sql_to_mongodb(&query.content) {
            Ok(operation) => {
                match self.execute_mongodb_operation(operation).await {
                    Ok(data) => {
                        let execution_time = start_time.elapsed().as_millis() as f64;
                        let rows_processed = if data.is_array() {
                            data.as_array().unwrap().len() as u64
                        } else {
                            1
                        };

                        Ok(DataSourceResult {
                            success: true,
                            data,
                            error: None,
                            stats: DataSourceStats {
                                execution_time_ms: execution_time,
                                rows_processed,
                                bytes_processed: 0, // 简化处理
                            },
                        })
                    }
                    Err(e) => {
                        Ok(DataSourceResult {
                            success: false,
                            data: Value::Null,
                            error: Some(e.to_string()),
                            stats: DataSourceStats {
                                execution_time_ms: start_time.elapsed().as_millis() as f64,
                                rows_processed: 0,
                                bytes_processed: 0,
                            },
                        })
                    }
                }
            }
            Err(e) => {
                Ok(DataSourceResult {
                    success: false,
                    data: Value::Null,
                    error: Some(format!("SQL parsing error: {}", e)),
                    stats: DataSourceStats {
                        execution_time_ms: start_time.elapsed().as_millis() as f64,
                        rows_processed: 0,
                        bytes_processed: 0,
                    },
                })
            }
        }
    }

    async fn health_check(&self) -> Result<Value> {
        if let Some(database) = &self.database {
            match database.run_command(doc! {"ping": 1}).await {
                Ok(_) => Ok(json!({
                    "status": "healthy",
                    "database": self.config.database_name,
                    "connection_string": self.config.connection_string
                })),
                Err(e) => Ok(json!({
                    "status": "unhealthy",
                    "error": e.to_string()
                })),
            }
        } else {
            Ok(json!({
                "status": "disconnected",
                "message": "MongoDB client not connected"
            }))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_mongodb_config_default() {
        let config = MongoDBConfig::default();
        assert_eq!(config.connection_string, "mongodb://localhost:27017");
        assert_eq!(config.database_name, "test");
        assert_eq!(config.connect_timeout_seconds, 30);
        assert_eq!(config.query_timeout_seconds, 60);
        assert_eq!(config.max_connections, 10);
    }

    #[test]
    fn test_mongodb_data_source_creation() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        assert_eq!(data_source.name(), "MongoDB");
        assert_eq!(data_source.description(), "MongoDB文档数据库数据源");
        assert!(!data_source.connected);
    }

    #[test]
    fn test_parse_simple_select_query() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        let sql = "SELECT * FROM users";
        let operation = data_source.parse_sql_to_mongodb(sql).unwrap();

        match operation {
            MongoDBOperation::Find { collection, filter, limit } => {
                assert_eq!(collection, "users");
                assert!(filter.is_empty());
                assert!(limit.is_none());
            }
            _ => panic!("Expected Find operation"),
        }
    }

    #[test]
    fn test_parse_select_with_where() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        let sql = "SELECT * FROM users WHERE id = 1";
        let operation = data_source.parse_sql_to_mongodb(sql).unwrap();

        match operation {
            MongoDBOperation::Find { collection, filter, limit } => {
                assert_eq!(collection, "users");
                assert!(!filter.is_empty());
                assert!(limit.is_none());
            }
            _ => panic!("Expected Find operation"),
        }
    }

    #[test]
    fn test_parse_select_with_limit() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        let sql = "SELECT * FROM users LIMIT 10";
        let operation = data_source.parse_sql_to_mongodb(sql).unwrap();

        match operation {
            MongoDBOperation::Find { collection, filter, limit } => {
                assert_eq!(collection, "users");
                assert!(filter.is_empty());
                assert_eq!(limit, Some(10));
            }
            _ => panic!("Expected Find operation"),
        }
    }

    #[test]
    fn test_parse_simple_condition() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        // 测试整数条件
        let condition = "id = 123";
        let result = data_source.parse_simple_condition(condition);
        assert!(result.is_some());
        let (field, value) = result.unwrap();
        assert_eq!(field, "id");
        assert_eq!(value, Bson::Int32(123));

        // 测试字符串条件
        let condition = "name = 'test'";
        let result = data_source.parse_simple_condition(condition);
        assert!(result.is_some());
        let (field, value) = result.unwrap();
        assert_eq!(field, "name");
        assert_eq!(value, Bson::String("test".to_string()));
    }

    #[test]
    fn test_parse_insert_query() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        let sql = "INSERT INTO users (name, email) VALUES ('test', 'test@example.com')";
        let operation = data_source.parse_sql_to_mongodb(sql).unwrap();

        match operation {
            MongoDBOperation::Insert { collection, .. } => {
                assert_eq!(collection, "default"); // 简化实现
            }
            _ => panic!("Expected Insert operation"),
        }
    }

    #[test]
    fn test_bson_to_json_conversion() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        // 测试字符串转换
        let bson_str = Bson::String("test".to_string());
        let json_val = data_source.bson_to_json(&bson_str).unwrap();
        assert_eq!(json_val, json!("test"));

        // 测试整数转换
        let bson_int = Bson::Int32(42);
        let json_val = data_source.bson_to_json(&bson_int).unwrap();
        assert_eq!(json_val, json!(42));

        // 测试布尔转换
        let bson_bool = Bson::Boolean(true);
        let json_val = data_source.bson_to_json(&bson_bool).unwrap();
        assert_eq!(json_val, json!(true));

        // 测试null转换
        let bson_null = Bson::Null;
        let json_val = data_source.bson_to_json(&bson_null).unwrap();
        assert_eq!(json_val, Value::Null);
    }

    #[tokio::test]
    async fn test_mongodb_data_source_not_connected() {
        let config = MongoDBConfig::default();
        let mut data_source = MongoDBDataSource::new(config);

        assert!(!data_source.is_connected().await);

        let query = DataSourceQuery {
            id: "test_query".to_string(),
            content: "SELECT * FROM users".to_string(),
            parameters: HashMap::new(),
        };

        let result = data_source.execute_query(query).await.unwrap();
        assert!(!result.success);
        assert!(result.error.is_some());
        assert!(result.error.unwrap().contains("not connected"));
    }

    #[tokio::test]
    async fn test_mongodb_health_check_disconnected() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        let health = data_source.health_check().await.unwrap();
        assert_eq!(health["status"], "disconnected");
    }

    #[test]
    fn test_unsupported_sql_operation() {
        let config = MongoDBConfig::default();
        let data_source = MongoDBDataSource::new(config);

        let sql = "CREATE TABLE users (id INT, name VARCHAR(50))";
        let result = data_source.parse_sql_to_mongodb(sql);
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("Unsupported SQL operation"));
    }
}
